package com.bff.gaia.client.cli;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.concurrent.EventExecutorGroup;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
import com.bff.gaia.api.common.ExecutionConfig;
import com.bff.gaia.api.common.JobExecutionResult;
import com.bff.gaia.api.common.JobSubmissionResult;
import com.bff.gaia.api.common.Plan;
import com.bff.gaia.api.common.accumulators.AccumulatorHelper;
import com.bff.gaia.api.java.operators.DataSink;
import com.bff.gaia.client.deployment.ClusterDescriptor;
import com.bff.gaia.client.deployment.ClusterSpecification;
import com.bff.gaia.client.program.*;
import com.bff.gaia.configuration.Configuration;
import com.bff.gaia.configuration.CoreOptions;
import com.bff.gaia.runtime.jobgraph.JobGraph;
import com.bff.gaia.util.GaiaException;
import com.bff.gaia.util.Preconditions;
import com.bff.gaia.util.ShutdownHookUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.duration.FiniteDuration;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import static com.bff.gaia.client.program.ClusterClient.MAX_SLOTS_UNKNOWN;

public class ServerHandlers extends ChannelInboundHandlerAdapter {
	public static boolean[] visited=new boolean[100];
	private int byteRead;
	private volatile int start = 0;
	int flag=0;
	private String file_dir ;
	private java.util.logging.Logger log= (java.util.logging.Logger) java.util.logging.Logger.getLogger(String.valueOf(ServerHandlers.class));
	private  Configuration config;
	private  int defaultParallelism;
	private final List<CustomCommandLine<?>> customCommandLines;
	private static final Logger LOG = LoggerFactory.getLogger(CliFrontend.class);
	private final Options customCommandLineOptions = new Options();
	public static ArrayList<ClusterClient> cliList=new ArrayList<>();
	public static ArrayList<PackagedProgram> proList=new ArrayList<>();
	public static ArrayList<Integer> parallelList=new ArrayList<>();
	public static boolean isCombine=false;
	public static ArrayList<Plan> planList=new ArrayList<Plan>();
	public static ArrayList<JobWithJars> jarList=new ArrayList<JobWithJars>();
	public static ArrayList<Plan> exePlan=new ArrayList<>();
	public static ArrayList<JobWithJars> jarLists=new ArrayList<>();
	public static List<String> ts=new ArrayList<String>();
	public static boolean isRun=false;
	public static ArrayList<String> args=new ArrayList<>();
	public static int count=0;


	public static  ArrayList<Plan> todoPlan=new ArrayList<>();
	public static ArrayList<JobWithJars> todojars=new ArrayList<>();


	public ServerHandlers(Configuration con,List<CustomCommandLine<?>> customCommandLines)
	{
		String str=System.getProperty("user.dir");
		file_dir=str;
		String stt=file_dir+"\\testsProgram";
		File dir=new File(stt);
		if(!dir.exists())
			dir.mkdir();
		file_dir=stt;
		System.out.println("文件存放的位置为："+file_dir);
		this.customCommandLines=customCommandLines;
		this.config=con;
		this.defaultParallelism=config.getInteger(CoreOptions.DEFAULT_PARALLELISM);
	}
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		// TODO Auto-generated method stub
		super.channelActive(ctx);
		log.info("服务端：channelActive()");
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		// TODO Auto-generated method stub
		//super.channelInactive(ctx);
		log.info("服务端：channelInactive()");
		ctx.flush();

		ctx.close();
	}
	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		if (msg instanceof FileUploadFile) {
			/*String []args=((FileUploadFile) msg).getArgs();
			System.out.println("含有jar文件的参数为："+args[1].toString());*/
			FileUploadFile ef = (FileUploadFile) msg;
			byte[] bytes = ef.getBytes();
			byteRead = ef.getEndPos();
			String md5 = ef.getFile_md5();//文件名
			String path = file_dir + File.separator + md5;
			File file = new File(path);
			RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw");
			randomAccessFile.seek(start);
			randomAccessFile.write(bytes);

			start = start + byteRead;
			System.out.println("path:"+path+","+byteRead);
			if (byteRead > 0) {
				ctx.writeAndFlush(start);
				randomAccessFile.close();
				if(byteRead!=1024 * 5){
					Thread.sleep(1000);
					channelInactive(ctx);
				}
			} else {
				//ctx.flush();
				ctx.close();
			}
			/*ByteBuf buf=(ByteBuf)msg;
			byte [] data=new byte[buf.readableBytes()];
			buf.readBytes(data);
			System.out.println(new String(data));
			String da=new String(data);
			System.out.println("服务端收到的数据为："+da);*/
			String[] split = path.split(" ");
			String argsii="";
			argsii+=path;
			if(((FileUploadFile) msg).getMainClass()!=null)
			{
				argsii+=" ";
				argsii+= ((FileUploadFile) msg).getMainClass();
			}
			if(((FileUploadFile) msg).getInput()!=null)
			{
				argsii+=" ";
				argsii+="--input";
				argsii+=" ";
				argsii+= ((FileUploadFile) msg).getInput();
			}
			if(((FileUploadFile) msg).getOutput()!=null)
			{
				argsii+=" ";
				argsii+="--output";
				argsii+=" ";
				argsii+= ((FileUploadFile) msg).getOutput();
			}
			int fg=0;

			for(int i=0; i<args.size(); i++)
			{
				//System.out.println(args.get(i));
				if(args.get(i).equals(argsii))
				{
					System.out.println("args中已经包含了这个参数命令行：");
					fg=1;
					break;
				}
			}
			if(fg==0)
			{
				args.add(argsii);
				System.out.println("把这一轮的args参数值加入到队列中："+argsii);
			}
			int flagged=0;
			if(ts.contains(path) )
			{
				flagged=1;
			}
			if(flagged==0)
			{
				//NettyServers.connect_num-=1;
				ts.add(path);
			}
			for(int i=0; i<split.length; i++)
			{
				System.out.println(split[i]);
			}
			if(NettyServers.connect_num==0 || NettyServers.count_clinets>3)
			{
				/***
				 * 先探测一下args数组中有哪些元素，它们的值是否一致
				 */
				System.out.println("其中的参数命令行如下，为：");
				for(int i=0; i<args.size(); i++)
				{
					System.out.println(args.get(i));
				}
				for(int i=0; i<ts.size(); i++)
				{
					//System.out.println("经过一次运行之后， Plan全局变量中的Plan个数变为："+ServerHandlers.planList.size());
					Options commandOptions = CliFrontendParser.getRunCommandOptions();
					Options commandLineOptions = CliFrontendParser.mergeOptions(commandOptions, customCommandLineOptions);
					String[] ss = args.get(i).split(" ");
					/***
					 * 把input、output、-c等参数重新构建成为命令行
					 */

					final CommandLine commandLine = CliFrontendParser.parse(commandLineOptions, ss, true);
					final RunOptions runOptions = new RunOptions(commandLine);
					// evaluate help flag
					if (runOptions.isPrintHelp()) {
						CliFrontendParser.printHelpForRun(customCommandLines);
						return;
					}
					if (runOptions.getJarFilePath() == null) {
						throw new CliArgsException("The program JAR file was not specified.");
					}
					else
					{
						//System.out.println("Options' jarFilePath is :"+runOptions.getJarFilePath());
					}
					PackagedProgram program;
					try {
						//LOG.info("Building program from JAR file");
						program = buildProgram(runOptions);
					}
					catch (FileNotFoundException e) {
						throw new CliArgsException("Could not build the program from JAR file.", e);
					}
					final CustomCommandLine<?> customCommandLine = getActiveCustomCommandLine(commandLine);
					try {
						runProgram(customCommandLine, commandLine, runOptions, program);
					} finally {
						program.deleteExtractedLibraries();
					}
				}
				System.out.println("进入作业选择合并阶段");
			}
		}
		else if(msg instanceof  Integer)
		{
			//System.out.println("文件传输结束啦啦啦啦啦啦");
			NettyServers.connect_num-=1;
			NettyServers.count_clinets+=1;
			System.out.println("经过一次连接，这个大小变为："+NettyServers.count_clinets);
		}
	}

	private <T> void runProgram(
		CustomCommandLine<T> customCommandLine,
		CommandLine commandLine,
		RunOptions runOptions,
		PackagedProgram program) throws ProgramInvocationException, GaiaException {
		final ClusterDescriptor<T> clusterDescriptor = customCommandLine.createClusterDescriptor(commandLine);
		try {

			final T clusterId = customCommandLine.getClusterId(commandLine);
			final ClusterClient<T> client;
			// directly deploy the job if the cluster is started in job mode and detached
			if (clusterId == null && runOptions.getDetachedMode()) {

				int parallelism = runOptions.getParallelism() == -1 ? defaultParallelism : runOptions.getParallelism();
				final JobGraph jobGraph = PackagedProgramUtils.createJobGraph(program, this.config, parallelism);
				final ClusterSpecification clusterSpecification = customCommandLine.getClusterSpecification(commandLine);
				client = clusterDescriptor.deployJobCluster(
					clusterSpecification,
					jobGraph,
					runOptions.getDetachedMode());
				System.out.println("Job has been submitted with JobID " + jobGraph.getJobID());
				try {
					client.shutdown();
				} catch (Exception e) {
					//LOG.info("Could not properly shut down the client.", e);
				}
			} else {

				final Thread shutdownHook;
				if (clusterId != null) {

					client = clusterDescriptor.retrieve(clusterId);
					shutdownHook = null;
				} else {

					// also in job mode we have to deploy a session cluster because the job
					// might consist of multiple parts (e.g. when using collect)
					final ClusterSpecification clusterSpecification = customCommandLine.getClusterSpecification(commandLine);
					client = clusterDescriptor.deploySessionCluster(clusterSpecification);
					// if not running in detached mode, add a shutdown hook to shut down cluster if client exits
					// there's a race-condition here if cli is killed before shutdown hook is installed
					if (!runOptions.getDetachedMode() && runOptions.isShutdownOnAttachedExit()) {
						shutdownHook = ShutdownHookUtil.addShutdownHook(client::shutDownCluster, client.getClass().getSimpleName(), LOG);
					} else {
						shutdownHook = null;
					}
				}
				try {
					client.setPrintStatusDuringExecution(runOptions.getStdoutLogging());
					client.setDetached(runOptions.getDetachedMode());
					//LOG.debug("Client slots is set to {}", client.getMaxSlots());

					//LOG.debug("{}", runOptions.getSavepointRestoreSettings());

					int userParallelism = runOptions.getParallelism();
					//LOG.debug("User parallelism is set to {}", userParallelism);
					if (client.getMaxSlots() != MAX_SLOTS_UNKNOWN && userParallelism == -1) {
						System.out.println("Using the parallelism provided by the remote cluster ("
							+ client.getMaxSlots() + "). "
							+ "To use another parallelism, set it at the ./bin/gaia client.");
						userParallelism = client.getMaxSlots();
					} else if (ExecutionConfig.PARALLELISM_DEFAULT == userParallelism) {
						userParallelism = defaultParallelism;
					}
					ServerHandlers.cliList.add(client);
					ServerHandlers.proList.add(program);
					ServerHandlers.parallelList.add(userParallelism);

					/***
					 * 选择合适的作业合并
					 */
					if(ServerHandlers.exePlan.size()<4){
						executePrograms(program, client, userParallelism);
					}



				} finally {
					if (clusterId == null && !client.isDetached()) {
						// terminate the cluster only if we have started it before and if it's not detached
						try {
							//client.shutDownCluster();
						} catch (final Exception e) {
							//LOG.info("Could not properly terminate the Gaia cluster.", e);
						}
						if (shutdownHook != null) {
							// we do not need the hook anymore as we have just tried to shutdown the cluster.
							//	ShutdownHookUtil.removeShutdownHook(shutdownHook, client.getClass().getSimpleName(), LOG);
						}
					}
					try {
						//client.shutdown();
					} catch (Exception e) {
						//LOG.info("Could not properly shut down the client.", e);
					}
				}
			}
		} finally {
			try {
				//clusterDescriptor.close();
			} catch (Exception e) {
				//LOG.info("Could not properly close the cluster descriptor.", e);
			}
		}
	}
	private void executePrograms(PackagedProgram proList, ClusterClient cliList, Integer parallelList) throws ProgramMissingJobException, ProgramInvocationException {
		System.out.println("Starting execution of program");
		JobSubmissionResult result = cliList.run(proList, parallelList);
		if(result==null)
		{
			System.out.println("非并行运行的作业输出结果为空");
		}
		//System.out.println("执行作业的得到结果");
		if (null == result && !ServerHandlers.isRun)
		{
			throw new ProgramMissingJobException("No JobSubmissionResult returned, please make sure you called " +
				"ExecutionEnvironment.execute()");
		}
		if(null==result && ServerHandlers.isRun)
		{
			System.out.println("当前没到合并的作业数量");
			return;
		}
		else {
			if (result.isJobExecutionResult()) {
				/***
				 * 可以删除文件夹下面的所有文件
				 */

			}
			if (result.isJobExecutionResult()) {
				System.out.println("多作业的并行执行结果返回");

				/*********清空planList中的内容*********/
				clearArrayList();
				System.out.println("Program execution finished");
				JobExecutionResult execResult = result.getJobExecutionResult();
				System.out.println("Job with JobID " + execResult.getJobID() + " has finished.");
				System.out.println("Job Runtime: " + execResult.getNetRuntime() + " ms");
				Map<String, Object> accumulatorsResult = execResult.getAllAccumulatorResults();
				if (accumulatorsResult.size() > 0) {
					System.out.println("Accumulator Results: ");
					System.out.println(AccumulatorHelper.getResultsFormatted(accumulatorsResult));
				}
			} else {
				System.out.println("Job has been submitted with JobID " + result.getJobID());
			}
		}

	}
	private void clearArrayList() {
		NettyServers.count_clinets=0;
		NettyServers.connect_num=10;
		ServerHandlers.planList.clear();
		ServerHandlers.ts.clear();
		ServerHandlers.jarList.clear();
		ServerHandlers.exePlan.clear();
		ServerHandlers.args.clear();
		ServerHandlers.todojars.clear();
		System.out.println("待清理的文件夹名称为："+file_dir);
		//delFolder(file_dir);
		for(int i=0; i<ServerHandlers.visited.length; i++){
			/****
			 ****重置访问标识
			 */
			ServerHandlers.visited[i]=false;
		}
	}
	/***
	 * 删除指定文件夹下所有文件
	 *
	 * @param path 文件夹完整绝对路径
	 * @return
	 */
	public boolean delAllFile(String path) {
		boolean flag = false;
		File file = new File(path);
		if (!file.exists()) {
			return flag;
		}
		if (!file.isDirectory()) {
			return flag;
		}
		String[] tempList = file.list();
		File temp = null;
		for (int i = 0; i < tempList.length; i++) {
			if (path.endsWith(File.separator)) {
				temp = new File(path + tempList[i]);
			} else {
				temp = new File(path + File.separator + tempList[i]);
			}
			if (temp.isFile()) {
				temp.delete();
			}
			if (temp.isDirectory()) {
				delAllFile(path + "/" + tempList[i]);// 先删除文件夹里面的文件
				delFolder(path + "/" + tempList[i]);// 再删除空文件夹
				flag = true;
			}
		}
		return flag;
	}

	/***
	 * 删除文件夹
	 *
	 * @param
	 */
	public  void delFolder(String folderPath) {
		try {
			boolean ss=delAllFile(folderPath); // 删除完里面所有内容
			if(ss){
				System.out.println("删除文件夹成功");
			}
			String filePath = folderPath;
			filePath = filePath.toString();
			java.io.File myFilePath = new java.io.File(filePath);
			myFilePath.delete(); // 删除空文件夹
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	protected PackagedProgram buildProgram(ProgramOptions options)
		throws FileNotFoundException, ProgramInvocationException {
		String[] programArgs = options.getProgramArgs();
		String jarFilePath = options.getJarFilePath();
		List<URL> classpaths = options.getClasspaths();
		if(programArgs==null)
		{
			System.out.println("entryPointClass空");
		}
		if(jarFilePath==null)
		{
			System.out.println("jarFilePath空");
		}
		if(classpaths==null)
		{
			System.out.println("classpath空");
		}
		if (jarFilePath == null) {
			throw new IllegalArgumentException("The program JAR file was not specified.");
		}
		File jarFile = new File(jarFilePath);
		//System.exit(0);
		// Check if JAR file exists
		if (!jarFile.exists()) {
			throw new FileNotFoundException("JAR file does not exist: " + jarFile);
		}
		else if (!jarFile.isFile()) {
			throw new FileNotFoundException("JAR file is not a file: " + jarFile);
		}
		else
		{
			System.out.println("Jar 文件存在 ,其路径为："+jarFile.getAbsolutePath());
		}
		// Get assembler class
		String entryPointClass = options.getEntryPointClassName();
		PackagedProgram program = entryPointClass == null ? new PackagedProgram(jarFile, classpaths, programArgs) :
			new PackagedProgram(jarFile, classpaths, entryPointClass, programArgs);
		program.setSavepointRestoreSettings(options.getSavepointRestoreSettings());

		return program;
	}
	public CustomCommandLine<?> getActiveCustomCommandLine(CommandLine commandLine) {
		for (CustomCommandLine<?> cli : customCommandLines) {
			if (cli.isActive(commandLine)) {
				return cli;
			}
		}
		throw new IllegalStateException("No command-line ran.");
	}
	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		super.channelReadComplete(ctx);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		super.exceptionCaught(ctx, cause);
	}

}